package com.deltapunkt.proxy;


import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingDeque;

import com.deltapunkt.proxy.factory.ProxyFactory;

/* 
 * create new connections
 * handle IO read and create new work for processing threads
 * process IO write - ie work sent by the processing threads
 * 
 */

public class AsyncServer implements Runnable {
	private ByteBuffer buffer;
	private Selector selector;
	private BlockingDeque<Command> commandQueue;
	private ConcurrentHashMap<SocketChannel, Proxy> proxyMap;
	private ConcurrentHashMap<SocketChannel, BlockingDeque<ByteBuffer>> writeMap;
	// private Executor exec;
	private ProxyFactory proxyFactory;
	private TaskScheduler scheduler;

	public AsyncServer(TaskScheduler scheduler, ProxyFactory proxyFactory,
			SocketAddress srcAddress) throws IOException {
		this.scheduler = scheduler;
		this.proxyFactory = proxyFactory;
		commandQueue = new LinkedBlockingDeque<Command>();
		proxyMap = new ConcurrentHashMap<SocketChannel, Proxy>();
		writeMap = new ConcurrentHashMap<SocketChannel, BlockingDeque<ByteBuffer>>();
		buffer = ByteBuffer.wrap(new byte[8192]);
		// exec = Executors.newFixedThreadPool(10); // maximum number of
		// concurrent connections
		selector = Selector.open();
		addListener(srcAddress);
	}

	public void addListener(SocketAddress addr) {
		try {
			ServerSocketChannel ssc = ServerSocketChannel.open();
//			InetSocketAddress isa = new InetSocketAddress(port);
			ssc.socket().bind(addr);

			registerChannel(ssc, SelectionKey.OP_ACCEPT);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void run() {
		try {
			while (true) {
				for (Command c : commandQueue) {
					processCommand(c);
				}
				commandQueue.clear();

				if (selector.select() > 0) {
					System.out.println(selector.selectedKeys());
				} else {
					System.out.println("[nothing selected ...]");
					continue;
				}

				for (Iterator<SelectionKey> it = selector.selectedKeys()
						.iterator(); it.hasNext();) {
					SelectionKey key = it.next();
					it.remove();

					if (!key.isValid()) {
						continue;
					}

					if (key.isAcceptable()) {
						accept(key);
					} else if (key.isConnectable()) {
						connect(key);
					} else if (key.isReadable()) {
						read(key);
					} else if (key.isWritable()) {
						write(key);
					}
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	// target connect
	private void connect(SelectionKey key) {
		SocketChannel s = (SocketChannel) key.channel();

		try {
			s.finishConnect();
		} catch (IOException e) {
			e.printStackTrace();
			key.cancel();
			return;
		}
		Proxy proxy = proxyMap.get(s);
		proxy.read(new State("connected"));
		scheduler.addTask(proxy);
		// proxyMap.get(s).setTargetConnected();
	}

	// static final int OP_NONE = 0;

	private void accept(SelectionKey key) {
		SocketChannel s;
		try {
			ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
			s = ssc.accept();
			// register with no operation = zero = 0
			registerChannel(s, Command.OP_NONE);

			//use google guice to insert the appropriate proxy here
			Proxy proxy = proxyFactory.createProxy(this, s);

			mapSocket(s, proxy);
			scheduler.addTask(proxy); // to execute connect
			// exec.execute(proxy);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void mapSocket(SocketChannel s, Proxy asyncPortProxy) {
		writeMap.put(s, new LinkedBlockingDeque<ByteBuffer>());
		proxyMap.put(s, asyncPortProxy);
	}

	private void read(SelectionKey key) {
		// queue-driven flow
		// add to queue; if about to fill the queue, addCommand(s, OP_NONE)
		// so that the selector stops triggering a read
		// when reading from the queue, check whether the queue was full and
		// addCommand(s, OP_READ) as necessary
		SocketChannel s = (SocketChannel) key.channel();
		buffer.clear();
		int cnt = -1;
		try {
			cnt = s.read(buffer);
		} catch (IOException e) {
			e.printStackTrace();
			closeChannel(s);
		}

		if (cnt == -1) {
			closeChannel(s);
		}

		Proxy proxy = proxyMap.get(s);
		proxy.read(new Message(s, buffer));
		scheduler.addTask(proxy);
	}

	private void write(SelectionKey key) {
		SocketChannel s = (SocketChannel) key.channel();
		BlockingDeque<ByteBuffer> queue = writeMap.get(s);
		while (!queue.isEmpty()) {
			ByteBuffer buffer = queue.getFirst();
			try {
				s.write(buffer);
			} catch (IOException e) {
				e.printStackTrace();
			}
			if (buffer.remaining() > 0) {
				// expand this to retry
				break;
			}
			queue.removeFirst();
		}
		setChannelOps(s, SelectionKey.OP_READ);
	}

	private void processCommand(Command c) throws IOException {
		switch (c.getType()) {
		case REGISTER:
			registerChannel(c.getChannel(), c.getOps());
			break;
		case SETOPS:
			setChannelOps(c.getChannel(), c.getOps());
			break;
		case TERMINATE:
			closeChannel(c.getChannel());
			break;
		}
	}

	private void registerChannel(SelectableChannel s, int ops)
			throws IOException {
		s.configureBlocking(false);
		s.register(selector, ops);
	}

	private void setChannelOps(SocketChannel s, int ops) {
		SelectionKey key = s.keyFor(selector);
		key.interestOps(ops);
	}

	private void closeChannel(SocketChannel s) {
		s.keyFor(selector).cancel();
		try {
			s.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	// add message to the write queue
	public void addMessage(Message m) throws InterruptedException {
		writeMap.get(m.getChannel()).putLast(m.getContentAsBuffer());
	}

	public void addCommand(Command cmd) {
		commandQueue.add(cmd);
		selector.wakeup();
	}
}
